package com.sjc.lesson02.api.streaming.transformation;

import com.sjc.lesson02.api.domain.WordCount;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * 滑动窗口实现单词计数
 * 数据源：socket
 * 需求：每隔1秒计算最近2秒单词出现的次数
 * 练习算子：
 *  flatMap
 *  keyBy:
 *      dataStream.keyBy("someKey") // 指定对象中的 "someKey"字段作为分组key
 *      dataStream.keyBy(0) //指定Tuple中的第一个元素作为分组key
 *  sum
 */
public class WindowWordCountJava {
    public static void main(String[] args) throws Exception {
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        int port = parameterTool.getInt("port");
        String hostname = parameterTool.get("hostname");
        String delimiter = "\t";

        // 1. 获取flink运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 2. 获取数据源
        DataStreamSource<String> textStream = env.socketTextStream(hostname, port, delimiter);
        // 3. 执行transformation操作
        SingleOutputStreamOperator<WordCount> wordCountStream = textStream.flatMap(new FlatMapFunction<String, WordCount>() {
            @Override
            public void flatMap(String line, Collector<WordCount> collector) throws Exception {
                String[] fileds = line.split("\t");
                for (String word : fileds) {
                    collector.collect(new WordCount(word, 1l));
                }
            }
        }).keyBy("word")
                .timeWindow(Time.seconds(2), Time.seconds(1)) // 每隔1秒计算最近2秒的数据
                .sum("count");

        wordCountStream.print().setParallelism(1);
        // 4. 运行程序
        env.execute("WindowWordCountJava");
    }
}
